package com.example.springbootelasticsearch.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.Resource;
import java.util.List;

/**
 * @Description: 发布者：监听到 mysql数据变化后，发送广播事件；一次广播可以有多个监听者，比如缓存更新、es更新、日志记录等
 * @Author zhangdongkang
 * @Date 2022/9/27
 */
@Slf4j
@Configuration
@ConditionalOnExpression("${canal.enabled}")
public class CanalPublisher implements Runnable, ApplicationContextAware {

	private ApplicationContext applicationContext;

	@Resource
	private ApplicationEventPublisher publisher;
	@Resource
	private CanalConnector canalConnector;

	@Override
	@Scheduled(fixedDelay = 1000)
	public void run() {
		long batchId = -1;
		try {
			int batchSize = 1000;
			Message message = canalConnector.getWithoutAck(batchSize);
			batchId = message.getId();
			List<CanalEntry.Entry> entries = message.getEntries();
			if (batchId != -1 && entries.size() > 0) {
				entries.forEach(entry -> {
					if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
						// 解析处理
						publishCanalEvent(entry);
					}
				});
			}
			//提交确认
			canalConnector.ack(batchId);
		} catch (Exception e) {
			log.error("### canal client 解析数据异常:{}", e.getMessage());
			// 处理失败, 回滚数据
			canalConnector.rollback(batchId);
		}
	}

	/**
	 * 将 binlog 中的一条（entry），
	 * 解析成受影响的记录（change），再逐条解析受影响的记录（change），
	 * 将记录（rowData）的数据结构从 List 转成 Map，
	 * 完了交给 indexES 方式索引进 ElasticSearch；
	 *
	 * @param entry binlog 中的一条；
	 */
	private void publishCanalEvent(CanalEntry.Entry entry) {
		CanalEntry.EventType eventType = entry.getHeader().getEventType();
		CanalEntry.RowChange change = null;
		try {
			change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
		} catch (InvalidProtocolBufferException e) {
			log.error("### canal client 解析InvalidProtocolBufferException:{}", e.getMessage());
			return;
		}
		String database = entry.getHeader().getSchemaName();
		String table = entry.getHeader().getTableName();
		change.getRowDatasList().forEach(rowData -> {
			List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
			CanalEntry.Column idColumn = columns.stream().filter(column -> column.getIsKey()).findFirst().orElse(null);
			log.info("### canal received mysql change --> database:{}，table:{}，eventType:{}，id:{}",
							database, table, eventType.getValueDescriptor(), idColumn == null ? null : idColumn.getValue());
			//发布事件
			publisher.publishEvent(new CanalEvent(database, table, eventType, idColumn, columns));
		});
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}

}
